package com.lingjtx.im.core.distribute;

import com.lingjtx.common.core.constant.Const;
import com.lingjtx.common.core.constant.MqConst;
import com.lingjtx.common.core.constant.RedisConfig;
import com.lingjtx.common.core.model.mq.ImMessageBody;
import com.lingjtx.common.core.util.StringUtil;
import com.lingjtx.common.redis.service.IRedisService;
import com.lingjtx.common.thread.api.WebClientApi;
import com.lingjtx.im.core.common.SessionManager;
import com.lingjtx.im.core.handle.ImMessageHandle;
import com.lingjtx.im.core.protocol.MessageProto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.List;

@Slf4j
@Service
public class GroupMessage implements ImMessageHandle {

    private final WebClientApi webClientApi;

    private final IRedisService redisService;

    private final MessagePublisher publisher;

    public GroupMessage(WebClientApi webClientApi, IRedisService redisService, MessagePublisher publisher) {
        this.webClientApi = webClientApi;
        this.redisService = redisService;
        this.publisher = publisher;
    }

    @Override
    public void process(MessageProto.IMMessage message, byte version) {
        // 持久化群消息
        publisher.publish(MqConst.IM_QUEUE_GROUP_CHAT, message);

        // 拉取群成员 ID 列表
        webClientApi.getGroupMemberIds(message.getReceiverId())
                .doOnError(error -> log.error("Failed to get group members for groupId={}: {}", message.getReceiverId(), error.getMessage(), error))
                .subscribe(members -> handleGroupMessageDispatch(members, message));
    }

    private void handleGroupMessageDispatch(List<Long> memberIds, MessageProto.IMMessage message) {
        if (StringUtil.isEmpty(memberIds)) {
            log.warn("Group {} has no members.", message.getReceiverId());
            return;
        }

        for (Long userId : memberIds) {
            if (userId.equals(message.getSenderId())) continue; // 跳过自己

            if (SessionManager.isUserOnline(userId, redisService)) {
                String nodeId = (String) redisService.get(RedisConfig.IM_USER_ONLINE + userId);
                if (nodeId != null) {
                    // 替换掉群聊消息的 groupId
                    MessageProto.IMMessage newMessage = MessageProto.IMMessage.newBuilder(message).setReceiverId(userId).build();
                    publisher.publish(MqConst.IM_QUEUE_NODE + nodeId, newMessage);
                } else {
                    log.warn("Online user {} has no nodeId mapping in Redis.", userId);
                }
            } else {
                ImMessageBody body = convertToPushBody(message);
                publisher.publish(MqConst.QUEUE_NOTIFICATION_PUSH, body);
            }
        }
    }

    private ImMessageBody convertToPushBody(MessageProto.IMMessage message) {
        ImMessageBody body = new ImMessageBody();
        body.setMsgId(message.getMsgId());
        body.setMsgType(message.getMsgType());
        body.setContent(message.getContent());
        body.setSenderId(message.getSenderId());
        body.setReceiverId(message.getReceiverId());
        body.setTimestamp(message.getTimestamp());
        return body;
    }

    @Override
    public byte getImType() {
        return Const.IMMsgType.GROUP_CHAT;
    }
}
